package com.example.demo;

/**
 * @Author lizhien (zhien.li@nascent.cn)
 * @Date 2020/8/4 19:51
 * @Version 1.0.0
 * @Description com.example.demo
 */

import java.text.SimpleDateFormat;
import java.util.Date;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Consumer implements Runnable {

    //消费者名称
    private String name;
    private MessageConsumer consumer;
    private Session session;
    private Connection connection;
    //设置发布地址为默认的URL地址
    private static final String BROKER_URL = "failover:(tcp://192.168.1.80:31616)?randomize=true";//ActiveMQConnection.DEFAULT_BROKER_URL;

    public Consumer(String name) throws JMSException {
        this.name = name;
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);

        connection = connectionFactory.createConnection();
        connection.start();
        //设置第一个参数为true，事务类型数据（应该可以进一步确保不重复消费）
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //设置每次处理的最大消息数为1
        Destination destination = session.createQueue("hello.queue?consumer.prefetchSize=1");

        consumer = session.createConsumer(destination);
    }

    public boolean receive() throws JMSException {

        //设置消息响应等待5s
        TextMessage msg = (TextMessage) consumer.receive(1000);
        if (msg == null) {
            System.out.println("消费者" + this.name + "消息队列为空！！！");
            return false;
        } else {
            SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String nowTime = df.format(new Date());
            System.out.println("消费者" + this.name + "在时刻:" + nowTime + "消费数据: 【" + msg.getText() + "】");
            return true;
        }

    }

    @Override
    public void run() {

        try {
            while (true) {
                this.receive();
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }

    public void close() throws JMSException {
        consumer.close();
        session.close();
        connection.close();
    }

    public static void main(String[] args) {

        try {

            //创建3个消费者线程
            Consumer consumerA = new Consumer("A");
            Consumer consumerB = new Consumer("B");
            Consumer consumerC = new Consumer("C");
            Consumer consumerD = new Consumer("D");
            Consumer consumerE = new Consumer("E");

            new Thread(consumerA).start();
            new Thread(consumerB).start();
            new Thread(consumerC).start();
            new Thread(consumerD).start();
            new Thread(consumerE).start();

        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}
